'use strict'

const assert = require('assert');
const BitField = require('bitfield')
const speedometer = require('speedometer')
const EventEmitter3 = require('eventemitter3');

const util = require('./util');
const logger = require('./logger')('Wire');
const PeerConnection = require('./peer-connect').PeerConnection;

const BITFIELD_GROW = 262144; // 最大文件尺寸 BITFIELD_GROW * PIECE_SIZE(256k) = 64G
const KEEP_ALIVE_TIMER = 55000;
const BITTORRENT_PROTOCOL = 'BitTorrent protocol';
const REQUIRE_TIMEOUT = 20000;  // 请求超时时间


class Request {
  constructor(piece, offset, length, callback) {
    this.piece = piece
    this.offset = offset
    this.length = length
    this.callback = callback
  }
}

/**
 * Wire(bittorrent 协议实现, 目前版本只实现了bittorrent部分功能, 未实现的部分有 Extended(20) Port(20))
 *    协议文档参考: https: //blog.csdn.net/li6322511/article/details/79001422
 *
 *    事件:
 *        'ready': PeerConn连接成功同时触发, 只触发一次
 *        'keep-alive': 收到远端节点'keep-alive'后触发
 *        'handshake': 收到远端节点'handshake'后触发, 附加参数('infoHash', 'peerId')
 *        'choke': 收到远端节点'choke'后触发
 *        'unchoke': 收到远端节点'unchoke'后触发
 *        'interested': 收到远端节点'interested'后触发, 附加参数('infoHash', 'peerId')
 *        'uninterested': 收到远端节点'uninterested'后触发, 附加参数('infoHash', 'peerId')
 *        'have': 收到远端节点'have'后触发, 附加参数'index'
 *        'bitfield': 收到远端节点'bitfield'后触发, 附加参数'bitfield'
 *        'request': 收到远端节点数据块请求后触发, 附加参数('index', 'offset', 'length', 'respond')
 *        'upload': 向远端节点上传数据时触发, 附加参数 'length'
 *        'piece': 从远端节点下载数据时触发, 附加参数('index', 'offset', 'buffer')
 *        'download': 从远端节点下载数据时触发, 附加参数 'length'
 *        'cancel': 收到远端节点取消某数据块请求后触发, 附加参数 ('index', 'offset', 'length')
 *        'unknownmessage': 收到远端未知协议的消息后触发, 附加参数 'buffer'
 *        'timeout': 本地向远端节点发送的 request 请求超时
 *        'error': 出现错误时触发,附加参数 'error'
 *        'close': 本地关闭/远端关闭后触发, 只触发一次
 */
class Wire extends EventEmitter3 {
  /**
   * @param {PeerConnection} peerConn
   */
  constructor(peerConn) {
    super();
    this.peerId = peerConn.peerId;
    this.peerConn = peerConn;

    this._keepAliveInterval = null;
    this._timer = null;
    this._timeoutTime = REQUIRE_TIMEOUT;

    this.amChoking = true;  // are we choking the peer?
    this.amInterested = false;  // are we interested in the peer?

    this.peerChoking = true;  // is the peer choking us?
    this.peerInterested = false;  // is the peer interested in us?
    this.peerBitField = new BitField(0, {
      grow: BITFIELD_GROW
    })

    this.requests = []; // 本地向远端的资源请求队列
    this.peerRequests = []; // 远端的资源请求队列

    this.uploaded = 0;
    this.downloaded = 0;
    this.uploadSpeed = speedometer();
    this.downloadSpeed = speedometer();

    this.isReady = this.peerConn.isOpened;
    if (this.isReady) this.emit('ready');
    this.isClosed = false;

    this.hasSendHandShake = false;
    this.hasReciveHandShake = false;

    this._addConnListener();
  }

  /**
   * 监听peerConn状态
   */
  _addConnListener() {
    // 主动与远端建立的链接
    this.peerConn.once('open', () => {
      this.isReady = true;
      this.emit('ready');
    });

    // data
    this.peerConn.on('data', data => {
      if (this.isClosed) return;
      // window.myData = data;

      let dataLength;
      if (this.hasReciveHandShake) {
        if (data.length < 4) {
          logger.debug(`bad data(buffer length less than 4), ignore.`);
          return;
        }
        dataLength = data.readUInt32BE() + 4;
        if (dataLength !== data.length) {
          logger.debug(`data length is not correct, except: ${dataLength}bytes got: ${data.length}bytes.`);
          return;
        }
        this._onMessage(data.slice(4));
      } else {
        // handShake包 (handShake 没有len字段)
        const pstrlen = data.readUInt8(0);
        const protocol = data.slice(1, pstrlen + 1);
        if (protocol.toString() !== BITTORRENT_PROTOCOL) {
          logger.debug('wire not speaking BitTorrent protocol (%s), close it.', protocol.toString());
          this.close();
          return;
        }
        const pos = 1 + pstrlen + 8
          , infoHashBuffer = data.slice(pos, pos + 20)
          , peerBuffer = data.slice(pos + 20, pos + 40)
        this._onHandshake(infoHashBuffer, peerBuffer);
      }
    });

    this.peerConn.once('error', err => {
      logger.debug(`${err.toString()}, close it.`);
      this.close();
    });

    this.peerConn.once('close', () => {
      this.close();
    });
  }

  // ---------------------- OUTGOING MESSAGES  ----------------------
  /**
   * Message: "handshake" <pstrlen><pstr><reserved><info_hash><peer_id>
   *    pstrlen:  pstr的长度， 该值固定为19
   *    pstr: BitTorrent协议的关键字， 即“ BitTorrent protocol”
   *    reserved: 占8字节， 用于扩展BT协议， 一般这8字节都设置为0。 有些BT软件对BT协议进行了某些扩展， 因此可能看到有些peer发来的握手消息这8个字节不全为0， 不过不必理会， 这不会影响正常的通信
   *    info_hash: 与发往Tracker的GET请求中的info_hash为同一个值， 长度固定为20字节
   *    peer_id: 请求端的peer_id
   */
  handshake(infoHash, selfPeerid) {
    const MESSAGE_PROTOCOL = Buffer.from(`\u0013${BITTORRENT_PROTOCOL}`)
      , reservedBuffer = Buffer.from([0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
      , infoHashBuffer = Buffer.from(infoHash, 'hex')
      , peerIdBuffer = Buffer.from(selfPeerid, 'hex');

    logger.debug('handshake infoHash=%s peerId=%s', infoHash, selfPeerid);
    this._push(Buffer.concat([MESSAGE_PROTOCOL, reservedBuffer, infoHashBuffer, peerIdBuffer]));
    this.hasSendHandShake = true;
  }

  /**
   * Message "keep-alive": <len=0000>
   */
  keepAlive() {
    const MESSAGE_KEEP_ALIVE = Buffer.from([0x00, 0x00, 0x00, 0x00]);

    logger.debug('keep-alive');
    this._push(MESSAGE_KEEP_ALIVE);
  }

  /**
   * Message "choke": <len=0001><id=0>
   */
  choke() {
    const MESSAGE_CHOKE = Buffer.from([0x00, 0x00, 0x00, 0x01, 0x00]);

    if (this.amChoking) return;
    this.amChoking = true;
    logger.debug('choke');
    while (this.peerRequests.length) {
      this.peerRequests.pop();
    }
    this._push(MESSAGE_CHOKE);
  }

  /**
   * Message "unchoke": <len=0001><id=1>
   */
  unchoke() {
    const MESSAGE_UNCHOKE = Buffer.from([0x00, 0x00, 0x00, 0x01, 0x01]);

    if (!this.amChoking) return
    this.amChoking = false;
    logger.debug('unchoke');
    this._push(MESSAGE_UNCHOKE);
  }

  /**
   * Message "interested": <len=0001><id=2>
   */
  interested() {
    const MESSAGE_INTERESTED = Buffer.from([0x00, 0x00, 0x00, 0x01, 0x02]);

    if (this.amInterested) return
    this.amInterested = true;
    logger.debug('interested');
    this._push(MESSAGE_INTERESTED);
  }

  /**
   * Message "uninterested": <len=0001><id=3>
   */
  uninterested() {
    const MESSAGE_UNINTERESTED = Buffer.from([0x00, 0x00, 0x00, 0x01, 0x03]);

    if (!this.amInterested) return
    this.amInterested = false;
    logger.debug('uninterested');
    this._push(MESSAGE_UNINTERESTED);
  }

  /**
   * Message "have": <len=0005><id=4><piece index>
   * @param  {number} index
   */
  have(index) {
    // logger.debug('have %d', index);
    this._message(4, [index], null);
  }

  /**
   * Message "bitfield": <len=0001+X><id=5><bitfield>
   * @param  {BitField|Buffer} bitfield
   */
  bitfield(bitfield) {
    logger.debug('bitfield');
    if (!Buffer.isBuffer(bitfield)) bitfield = bitfield.buffer;
    this._message(5, [], bitfield);
  }

  /**
   * Message "request": <len=0013><id=6><index><begin><length>
   * @param  {number}   index
   * @param  {number}   offset
   * @param  {number}   length
   * @param  {function} cb
   */
  request(index, offset, length, cb) {
    if (!cb) cb = util.noop;
    if (!this.isReady) return cb(new Error('wire is closed'));
    if (this.peerChoking) return cb(new Error('peer is choking'))

    // logger.debug('request index=%d offset=%d length=%d', index, offset, length);

    this.requests.push(new Request(index, offset, length, cb));
    this._updateTimeout();
    this._message(6, [index, offset, length], null);
  }

  /**
   * Message "piece": <len=0009+X><id=7><index><begin><block>
   * @param  {number} index
   * @param  {number} offset
   * @param  {Buffer} buffer
   */
  piece(index, offset, buffer) {
    // logger.debug('piece index=%d offset=%d', index, offset);
    this.uploaded += buffer.length;
    this.uploadSpeed(buffer.length);
    this.emit('upload', buffer.length);
    this._message(7, [index, offset], buffer);
  }

  /**
   * Message "cancel": <len=0013><id=8><index><begin><length>
   * @param  {number} index
   * @param  {number} offset
   * @param  {number} length
   */
  cancel(index, offset, length) {
    logger.debug('cancel index=%d offset=%d length=%d', index, offset, length);
    this._callback(
      this._pull(this.requests, index, offset, length),
      new Error('request was cancelled'),
      null
    )
    this._message(8, [index, offset, length], null);
  }

  /**
   * Send a message to the remote peer.
   */
  _message(id, numbers, data) {
    const dataLength = data ? data.length : 0
    let buffer = Buffer.allocUnsafe(5 + (4 * numbers.length))

    // 大端字节序
    buffer.writeUInt32BE(buffer.length + dataLength - 4, 0)
    buffer[4] = id
    for (let i = 0; i < numbers.length; i++) {
      buffer.writeUInt32BE(numbers[i], 5 + (4 * i))
    }

    if (data) {
      buffer = Buffer.concat([buffer, data]);
    }
    this._push(buffer);
  }

  _push(data) {
    if (!this.peerConn.isOpened) return;
    // return this.peerConn.send(data);
    return this.peerConn.write(data);
  }
  // ----------------------------- (end) -----------------------------

  // ---------------------- INCOMING MESSAGES  ----------------------

  _onHandshake(infoHashBuffer, peerIdBuffer) {
    const infoHash = infoHashBuffer.toString('hex')
      , peerId = peerIdBuffer.toString('hex');

    logger.debug(`got handshake infoHash=${infoHash} peerId=${peerId}`);
    this.hasReciveHandShake = true;
    this.emit('handshake', infoHash, peerId);
  }

  _onKeepAlive() {
    logger.debug('got keep-alive');
    this.emit('keep-alive');
  }

  _onChoke() {
    this.peerChoking = true;
    logger.debug('got choke');
    this.emit('choke');
    while (this.requests.length) {
      this._callback(this.requests.pop(), new Error('peer is choking'), null);
    }
  }

  _onUnchoke() {
    this.peerChoking = false;
    logger.debug('got unchoke');
    this.emit('unchoke');
  }

  _onInterested() {
    this.peerInterested = true;
    logger.debug('got interested');
    this.emit('interested');
  }

  _onUninterested() {
    this.peerInterested = false;
    logger.debug('got uninterested');
    this.emit('uninterested');
  }

  _onHave(index) {
    if (this.peerBitField.get(index)) return;
    // logger.debug(`got have ${index}`);

    this.peerBitField.set(index);
    this.emit('have', index);
  }

  _onBitField(buffer) {
    this.peerBitField = new BitField(buffer);
    logger.debug('got bitfield');
    this.emit('bitfield', this.peerBitField);
  }

  _onRequest(index, offset, length) {
    if (this.amChoking) return;

    // logger.debug('got request index=%d offset=%d length=%d', index, offset, length);

    const respond = (err, buffer) => {
      if (request !== this._pull(this.peerRequests, index, offset, length)) return;
      if (err) {
        logger.debug(`error satisfying request index: ${index} offset: ${offset} length: ${length} ${err.message}.`);
        return;
      }
      // 请求数据片发送给请求端
      this.piece(index, offset, buffer);
    }

    const request = new Request(index, offset, length, respond);
    this.peerRequests.push(request);
    this.emit('request', index, offset, length, respond);
  }

  _onPiece(index, offset, buffer) {
    // logger.debug('got piece index=%d offset=%d', index, offset);
    this._callback(this._pull(this.requests, index, offset, buffer.length), null, buffer);
    this.downloaded += buffer.length;
    this.downloadSpeed(buffer.length);
    this.emit('download', buffer.length);
    this.emit('piece', index, offset, buffer);
  }

  _onCancel(index, offset, length) {
    logger.debug('got cancel index=%d offset=%d length=%d', index, offset, length);
    this._pull(this.peerRequests, index, offset, length);
    this.emit('cancel', index, offset, length);
  }

  /**
   * 处理远端peer的message
   * @param {Buffer} buffer
   */
  _onMessage(buffer) {
    if (buffer.length === 0) return this._onKeepAlive();

    const id = buffer[0];
    switch (id) {
      case 0:
        return this._onChoke();
      case 1:
        return this._onUnchoke();
      case 2:
        return this._onInterested();
      case 3:
        return this._onUninterested();
      case 4:
        return this._onHave(buffer.readUInt32BE(1));
      case 5:
        return this._onBitField(buffer.slice(1));
      case 6:
        return this._onRequest(
          buffer.readUInt32BE(1),
          buffer.readUInt32BE(5),
          buffer.readUInt32BE(9)
        );
      case 7:
        return this._onPiece(
          buffer.readUInt32BE(1),
          buffer.readUInt32BE(5),
          buffer.slice(9)
        );
      case 8:
        return this._onCancel(
          buffer.readUInt32BE(1),
          buffer.readUInt32BE(5),
          buffer.readUInt32BE(9)
        );
      case 9:
        logger.warn('bittorrent-protocol "port" we not support!');
        return;
      case 20:
        logger.warn('bittorrent-protocol "extended" we not support!');
        return;
      default:
        logger.debug('got unknown message');
        return this.emit('unknownmessage', buffer);
    }
  }

  // ----------------------------- (end) -----------------------------

  _onTimeout() {
    logger.debug('request timed out!');
    this._callback(this.requests.shift(), new Error('request has timed out.'), null);
    this.emit('timeout');
  }

  _callback(request, err, buffer) {
    if (!request) return

    this._clearTimeout();

    if (!this.peerChoking && this.isReady) this._updateTimeout()
    request.callback(err, buffer)
  }

  _clearTimeout() {
    if (!this._timer) return;

    clearTimeout(this._timer);
    this._timer = null;
  }

  _updateTimeout() {
    if (!this._timeoutTime || !this.requests.length || this._timer) return;

    this._timer = setTimeout(() => {
      this._onTimeout();
    }, this._timeoutTime);
  }

  _pull(requests, piece, offset, length) {
    for (let i = 0; i < requests.length; i++) {
      const req = requests[i]
      if (req.piece === piece && req.offset === offset && req.length === length) {
        util.quickUnorderArrayRemove(requests, i)
        return req
      }
    }
    return null
  }

  setkeepAlive(bool) {
    this._stopKeepAlive();
    if (!bool) return;
    this._keepAliveInterval = setInterval(() => {
      this.keepAlive();
    }, KEEP_ALIVE_TIMER);
  }

  _stopKeepAlive() {
    if (!this._keepAliveInterval) return;

    clearInterval(this._keepAliveInterval);
    this._keepAliveInterval = null;
  }

  _clearUp() {
    this._clearTimeout();
    this._stopKeepAlive();
    if (this.peerConn.isOpened) {
      this.peerConn.close();
    }

    while (this.peerRequests.length) {
      this.peerRequests.pop();
    }

    while (this.requests.length) {
      this._callback(this.requests.pop(), new Error('wire was closed'), null);
    }
  }

  close() {
    if (this.isClosed) return;

    this._clearUp();
    this.isReady = false;
    this.isClosed = true;
    this.emit('close');
  }
}

module.exports.Wire = Wire;
